package cn.redragon.soa.util;


import cn.redragon.soa.common.constant.RedisCacheKey;
import cn.redragon.soa.model.dto.stock.StockReservedDTO;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
import javax.annotation.PostConstruct;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.ReturnType;
import org.springframework.data.redis.connection.lettuce.LettuceConverters;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

@Slf4j
@Component
public class StockUtil {

    private static final String ZERO = "0";
    @Autowired
    private RedisTemplate<String, String> template;
    private HashOperations<String, String, String> hashOps;
    @Value("${spring.cache.hGetAll-size-limit:5000}")
    private Long hGetAllSizeLimit;

    /**
     * If any product is out of stock, then return the stock insufficient error message, do nothing on redis;
     * else return 'SUCCESS'.
     *
     * @param brand          String
     * @param reservedStocks list of StockReservedDTO {spu, sku, qty}
     * @return "success" or 'sku which has insufficient stock.'
     */
    public String checkStock(String brand, List<StockReservedDTO> reservedStocks) {
        String scriptTemplate = "local products={} %1$s local qtys={} %2$s local oos='' for k,v in pairs(products) do "
            +
            "KEYS[%3$d] = v local remains=redis.call('HGET', KEYS[%3$d], k) if not(remains) or tonumber(remains)<tonumber(qtys[k]) then oos=oos..k..';' end end "
            + "if(oos~='') then return oos else return 'success' end";

        return processStockChange(brand, reservedStocks, scriptTemplate);
    }

    /**
     * If any product is out of stock, then return the stock insufficient error message, do nothing on redis;
     * else decrease the stocks of order lines and return 'SUCCESS'.
     *
     * @param brand          String
     * @param reservedStocks list of StockReservedDTO {spu, sku, qty}
     * @return "success" or 'sku which has insufficient stock.'
     */
    public String reserveStock(String brand, List<StockReservedDTO> reservedStocks) {
        String scriptTemplate = "local products={} %1$s local qtys={} %2$s local oos='' for k,v in pairs(products) do "
            +
            "KEYS[%3$d] = v local remains=redis.call('HGET', KEYS[%3$d], k) if not(remains) or tonumber(remains)<tonumber(qtys[k]) then oos=oos..k..';' end end "
            + "if(oos~='') then return oos else for k,v in pairs(products) do "
            +
            "KEYS[%3$d] = v redis.call('HINCRBY', KEYS[%3$d], k, 0-qtys[k]) redis.call('HINCRBY', KEYS[%3$d], k..':reserved', qtys[k]) end return 'success' end";

        return processStockChange(brand, reservedStocks, scriptTemplate);
    }

    /**
     * return reserved stocks to available stocks
     *
     * @param brand          String
     * @param reservedStocks list of StockReservedDTO {spu, sku, qty}
     * @return "success"
     */
    public String cancelStock(String brand, List<StockReservedDTO> reservedStocks) {
        String scriptTemplate = "local products={} %1$s local qtys={} %2$s local oos='' for k,v in pairs(products) do "
            + "KEYS[%3$d] = v local exists=redis.call('HEXISTS', KEYS[%3$d], k..':reserved') if (exists==0) then oos=oos..k..';' end end "
            + "if(oos~='') then return oos else for k,v in pairs(products) do "
            +
            "KEYS[%3$d] = v redis.call('HINCRBY', KEYS[%3$d], k, qtys[k]) redis.call('HINCRBY', KEYS[%3$d], k..':reserved', 0-qtys[k]) end return 'success' end";

        return processStockChange(brand, reservedStocks, scriptTemplate);
    }

    /**
     * import oms stock in full mode
     *
     * @param brand          String
     * @param reservedStocks list of StockReservedDTO {spu, sku, qty}
     * @return "success"
     */
    public String importStock(String brand, List<StockReservedDTO> reservedStocks) {
        String scriptTemplate = "local products={} %1$s local qtys={} %2$s for k,v in pairs(products) do "
            + "KEYS[%3$d] = v local reserved=redis.call('HGET', KEYS[%3$d], k..':reserved') if not(reserved) then reserved='0' end "
            + "local available=redis.call('HGET', KEYS[%3$d], k) if not(available) then available='0' end "
            + "redis.call('HINCRBY', KEYS[%3$d], k, qtys[k]-tonumber(reserved)-tonumber(available)) end return 'success'";

        return processStockChange(brand, reservedStocks, scriptTemplate);
    }

    /**
     * import oms stock in incremental mode
     *
     * @param brand          String
     * @param reservedStocks list of StockReservedDTO {spu, sku, qty}
     * @return "success"
     */
    public String importStockIncremental(String brand, List<StockReservedDTO> reservedStocks) {
        String scriptTemplate = "local products={} %1$s local qtys={} %2$s for k,v in pairs(products) do "
            + "KEYS[%3$d] = v local reserved=redis.call('HGET', KEYS[%3$d], k..':reserved') if not(reserved) then reserved='0' end "
            + "local available=redis.call('HGET', KEYS[%3$d], k) if not(available) then available='0' end "
            + "redis.call('HINCRBY', KEYS[%3$d], k, qtys[k]-tonumber(reserved)) end return 'success'";

        return processStockChange(brand, reservedStocks, scriptTemplate);
    }

    /**
     * release reserved stocks after order exported to OMS
     *
     * @param brand          String
     * @param reservedStocks list of StockReservedDTO {spu, sku, qty}
     * @return "success"
     */
    public String releaseStock(String brand, List<StockReservedDTO> reservedStocks) {
        String scriptTemplate = "local products={} %1$s local qtys={} %2$s local oos='' for k,v in pairs(products) do "
            + "KEYS[%3$d] = v local exists=redis.call('HEXISTS', KEYS[%3$d], k..':reserved') if (exists==0) then oos=oos..k..';' end end "
            + "if(oos~='') then return oos else for k,v in pairs(products) do "
            + "KEYS[%3$d] = v redis.call('HINCRBY', KEYS[%3$d], k..':reserved', 0-qtys[k]) end return 'success' end";

        return processStockChange(brand, reservedStocks, scriptTemplate);
    }

    /**
     * This method is dedicated for value type = Long
     *
     * @param key   String
     * @param field String
     * @return Long
     */
    public Long hashGetLong(String key, String field) {
        return Long.valueOf(Optional.ofNullable(this.hashOps.get(key, field)).orElse(ZERO));
    }

    public Long hashIncrement(String key, String field, Long val) {
        return this.hashOps.increment(key, field, val);
    }

    public void hashSet(String key, String field, Object value) {
        this.hashOps.put(key, field, value.toString());
    }

    @Deprecated
    public Map<String, String> hashGetAll(String key) {
        Long size = hashOps.size(key);
        if (size > hGetAllSizeLimit) {
            log.warn("Switch to hashScan automatically since it's forbidden to use hGetAll on big map with size={} for key={}", size, key);
            return this.hashScan(key);
        }
        return this.hashOps.entries(key);
    }

    public Map<String, String> hashScan(String key) {
        Map<String, String> result = new HashMap<>();
        ScanOptions options = ScanOptions.scanOptions().count(hGetAllSizeLimit).build();
        Cursor<Map.Entry<String, String>> cursor = this.hashOps.scan(key, options);
        Map.Entry<String, String> entry;
        while (cursor.hasNext()) {
            entry = cursor.next();
            result.put(entry.getKey(), entry.getValue());
        }
        return result;
    }

    public boolean selectAndSetEx(Integer database, String key, String value, long timeout) {
        StringBuilder script = new StringBuilder();
        script.append("redis.call('select', ARGV[3]); return redis.call('SETEX', KEYS[1], ARGV[2], ARGV[1])");
        return this.template.execute((RedisConnection connection) -> connection
            .eval(script.toString().getBytes(), ReturnType.BOOLEAN, 1, key.getBytes(),
                  value.getBytes(), LettuceConverters.toBytes(timeout), LettuceConverters.toBytes(database)));
    }

    public boolean selectAndDel(Integer database, String key, String value) {
        StringBuilder script = new StringBuilder();
        script.append("redis.call('select', ARGV[2]); if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('DEL', KEYS[1]) else return 0 end");
        return this.template.execute((RedisConnection connection) -> connection
            .eval(script.toString().getBytes(), ReturnType.BOOLEAN, 1, key.getBytes(),
                  value.getBytes(), LettuceConverters.toBytes(database)));
    }

    @PostConstruct
    private void init() {
        this.hashOps = this.template.opsForHash();
    }

    private String processStockChange(final String brand, final List<StockReservedDTO> reservedStocks, final String scriptTemplate) {
        if (CollectionUtils.isEmpty(reservedStocks)) {
            throw new IllegalArgumentException("List reservedStocks is empty.");
        }
        final int size = reservedStocks.size();
        List<String> keyList = Lists.newArrayList();
        List<String> fieldList = Lists.newArrayList();
        Set<String> typeSet = Sets.newHashSet();
        List<Long> qtyList = Lists.newArrayList();
        StringBuilder productBuffer = new StringBuilder();
        StringBuilder qtyBuffer = new StringBuilder();
        final AtomicInteger fieldPointer = new AtomicInteger(0);

        reservedStocks.stream().forEach(i -> {
            // CR: change stock key from spu to sku
            //keyList.add(String.format(RedisCacheKey.SPU_STOCK, brand, i.getSku()));
            keyList.add(i.getSku());
            fieldList.add(i.getSku());
            qtyList.add(i.getQty());
            if (i.getType() == null) {
                throw new IllegalArgumentException("Stock type is mandatory.");
            }
            typeSet.add(i.getType());
            productBuffer.append("products[").append("KEYS[").append(fieldPointer.incrementAndGet()).append("]").append("]=")
                .append("ARGV[").append(fieldPointer.get()).append("] ");
            qtyBuffer.append("qtys[").append("KEYS[").append(fieldPointer.get()).append("]").append("]=")
                .append("ARGV[").append(size + fieldPointer.get()).append("] ");
        });

        if (typeSet.size() > 1) {
            throw new IllegalArgumentException("Stock type must be same.");
        }

        String type = typeSet.iterator().next();

        String script = String.format(scriptTemplate, productBuffer.toString(), qtyBuffer.toString(), fieldPointer.incrementAndGet());

        byte[][] keysAndArgs = new byte[size * 3][];
        for (int i = 0, keySize = fieldList.size(); i < keySize; i++) {
            keysAndArgs[i] = LettuceConverters.toBytes(String.format(RedisCacheKey.SKU_STOCK, brand, type, fieldList.get(i)));
        }

        final AtomicInteger keyPointer = new AtomicInteger(size);
        // after CR, actually keyList stores sku, not spu
        for (String spu : keyList) {
            keysAndArgs[keyPointer.getAndIncrement()] = LettuceConverters.toBytes(String.format(RedisCacheKey.SKU_STOCK, brand, type, spu));
        }

        for (Long qty : qtyList) {
            keysAndArgs[keyPointer.getAndIncrement()] = LettuceConverters.toBytes(qty);
        }

        return new String(template.execute((RedisCallback<byte[]>) con -> con.eval(script.getBytes(),
                                                                                   ReturnType.VALUE, fieldList.size(), keysAndArgs)));
    }
}